package com.tomcat.mesos.tomcat.scheduler;

import org.apache.mesos.Protos;
import org.apache.mesos.Scheduler;
import org.apache.mesos.SchedulerDriver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author Rayn on 2015/12/23 15:19
 * @email liuwei412552703@163.com
 */
public class TomcatScheduler implements Scheduler {
    private static final Logger LOG = LoggerFactory.getLogger(TomcatScheduler.class);


    /**
     * 生成Task ID .
     */
    private final AtomicInteger taskIDGenerator = new AtomicInteger();

    /**
     * 向 Mesos Master 注册
     *
     * @param driver
     * @param frameworkId
     * @param masterInfo
     */
    @Override
    public void registered(SchedulerDriver driver, Protos.FrameworkID frameworkId, Protos.MasterInfo masterInfo) {
        LOG.info("向 Mesos Master 注册.");


    }

    /**
     * 向备选 Master 进行注册
     *
     * @param driver
     * @param masterInfo
     */
    @Override
    public void reregistered(SchedulerDriver driver, Protos.MasterInfo masterInfo) {
        LOG.info("向 Mesos Master 注册.");


    }

    /**
     * 向框架提供资源 offer 时调用。每一份资源 offer 包含从某个 slave 上获取的资源列表。
     *
     * @param driver
     * @param offers
     */
    @Override
    public void resourceOffers(SchedulerDriver driver, List<Protos.Offer> offers) {
        for (Protos.Offer offer : offers) {
            Protos.TaskID taskID = Protos.TaskID.newBuilder().setValue(taskIDGenerator.incrementAndGet()).build();
            System.out.println("Launching task " + taskIDGenerator.get() + " on slave " + offer.getSlaveId().getValue());

            Protos.ExecutorInfo executor = Protos.ExecutorInfo.newBuilder()
                    .setExecutorId(Protos.ExecutorID.newBuilder().setValue(String.valueOf(tasksSubmitted)))
                    .setCommand(createCommand(task))
                    .setName("Tomcat Executor (Java)")
                    .build();

            Protos.TaskInfo taskInfo = Protos.TaskInfo.newBuilder()
                    .setName("TomcatTask-" + taskIDGenerator.get())
                    .setTaskId(taskID)
                    .setExecutor(Protos.ExecutorInfo.newBuilder(executor))
                    .addResources(Protos.Resource.newBuilder()
                        .setName("cpus")
                        .setType(Protos.Value.Type.SCALAR)
                        .setScalar(Protos.Value.Scalar.newBuilder().setValue(1)))

                    .addResources(Protos.Resource.newBuilder()
                        .setName("mem")
                        .setType(Protos.Value.Type.SCALAR)
                        .setScalar(Protos.Value.Scalar.newBuilder().setValue(128)))
                    .setSlaveId(offer.getSlaveId())
                    .build();

            driver.launchTasks(Collections.singletonList(offer.getId()), Collections.singletonList(taskInfo));
        }
    }

    /**
     *
     * @param driver
     * @param offerId
     */
    @Override
    public void offerRescinded(SchedulerDriver driver, Protos.OfferID offerId) {

    }

    /**
     * 任务调度状态更新
     *
     * @param driver
     * @param status
     */
    @Override
    public void statusUpdate(SchedulerDriver driver, Protos.TaskStatus status) {
        LOG.info("状态更新，task ID： " + status.getTaskId().getValue() + " ，状态为： " + status.getState());


        switch (status.getState()) {
            case TASK_RUNNING:
                pendingInstances.remove(taskId);
                runningInstances.add(taskId);
                break;
            case TASK_FAILED:
            case TASK_FINISHED:
                pendingInstances.remove(taskId);
                runningInstances.remove(taskId);
                break;
            default:
                break;
        }

        if (status.getState().equals(Protos.TaskState.TASK_FINISHED)) {
            LOG.info("Task " + status.getTaskId().getValue() + " 状态为 : Finished.");
        } else {
            LOG.info("Task ID: " + status.getTaskId().getValue() + ", 状态消息：" + status.getMessage());
        }

    }

    /**
     *
     * @param driver
     * @param executorId
     * @param slaveId
     * @param data
     */
    @Override
    public void frameworkMessage(SchedulerDriver driver, Protos.ExecutorID executorId, Protos.SlaveID slaveId, byte[] data) {

    }

    /**
     *
     * @param driver
     */
    @Override
    public void disconnected(SchedulerDriver driver) {

    }

    /**
     *
     * @param driver
     * @param slaveId
     */
    @Override
    public void slaveLost(SchedulerDriver driver, Protos.SlaveID slaveId) {

    }

    /**
     *
     * @param driver
     * @param executorId
     * @param slaveId
     * @param status
     */
    @Override
    public void executorLost(SchedulerDriver driver, Protos.ExecutorID executorId, Protos.SlaveID slaveId, int status) {

    }

    /**
     *
     * @param driver
     * @param message
     */
    @Override
    public void error(SchedulerDriver driver, String message) {

    }

    /**
     * 构建启动命令
     *
     * @param args
     * @return
     */
    private Protos.CommandInfo.Builder createCommand(String args) {

        String command = "java -cp tomcat-on-meos -Djava.library.path=/usr/local/lib/ com.tomcat.mesos.tomcat.executor.TomcatExecutor ";

        return Protos.CommandInfo.newBuilder().setValue(command + args);
    }
}
